package com.zetyun.tiger.connector.redis;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zetyun.rt.sdk.model.RtEvent;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.Objects;

@AllArgsConstructor
public class RtEventRedisMapper implements RedisMapper<RtEvent> {


    private RedisCommand redisCommand;

    private String additionalKey;

    private String keyName;

    private String valueName;

    private ObjectMapper om;

    @Override
    public RedisCommandDescription getCommandDescription() {

        return new RedisCommandDescription(redisCommand, additionalKey);
    }

    @Override
    public String getKeyFromData(RtEvent data) {
        return data.getField(keyName,String.class).trim();
    }

    @SneakyThrows
    @Override
    public String getValueFromData(RtEvent data) {
        if (Objects.isNull(valueName) || valueName.isEmpty()) {
            return om.writeValueAsString(data.getFields());
        }
        return data.getField(valueName,String.class).trim();
    }


}
